
package me.spring.cloud.common.components.buid.impl;

import me.spring.cloud.common.components.buid.BitsAllocator;
import me.spring.cloud.common.components.buid.buffer.BufferPaddingExecutor;
import me.spring.cloud.common.components.buid.buffer.BufferedUidProvider;
import me.spring.cloud.common.components.buid.buffer.RejectedPutBufferHandler;
import me.spring.cloud.common.components.buid.buffer.RejectedTakeBufferHandler;
import me.spring.cloud.common.components.buid.buffer.RingBuffer;
import me.spring.cloud.common.components.buid.exception.UidGenerateException;
import me.spring.cloud.common.components.util.lang.AssertUtil;
import java.util.ArrayList;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;

@Slf4j
public class CachedUidGenerator extends DefaultUidGenerator implements DisposableBean {

  private static final int DEFAULT_BOOST_POWER = 3;

  /** Spring properties */
  private int boostPower = DEFAULT_BOOST_POWER;
  private int paddingFactor = RingBuffer.DEFAULT_PADDING_PERCENT;
  private Long scheduleInterval;

  private RejectedPutBufferHandler rejectedPutBufferHandler;
  private RejectedTakeBufferHandler rejectedTakeBufferHandler;

  /** RingBuffer */
  private RingBuffer ringBuffer;
  private BufferPaddingExecutor bufferPaddingExecutor;

  @Override
  public void afterPropertiesSet() throws Exception {
    // initialize workerId & bitsAllocator
    super.afterPropertiesSet();

    // initialize RingBuffer & RingBufferPaddingExecutor
    this.initRingBuffer();
    log.info("Initialized RingBuffer successfully.");
  }

  @Override
  public long getUID() {
    try {
      return ringBuffer.take();
    } catch (Exception e) {
      log.error("Generate unique id exception. ", e);
      throw new UidGenerateException(e);
    }
  }

  @Override
  public String parseUID(long uid) {
    return super.parseUID(uid);
  }

  @Override
  public void destroy() throws Exception {
    bufferPaddingExecutor.shutdown();
  }

  /**
   * Get the UIDs in the same specified second under the max sequence
   *
   * @param currentSecond
   * @Return UID list, size of {@link BitsAllocator#getMaxSequence()} + 1
   */
  protected List<Long> nextIdsForOneSecond(long currentSecond) {
    // Initialize result list size of (max sequence + 1)
    int listSize = (int) bitsAllocator.getMaxSequence() + 1;
    List<Long> uidList = new ArrayList(listSize);

    // Allocate the first sequence of the second, the others can be calculated with the offset
    long firstSeqUid = bitsAllocator.allocate(currentSecond - epochSeconds, workerId, 0L);
    for (int offset = 0; offset < listSize; offset++) {
      uidList.add(firstSeqUid + offset);
    }

    return uidList;
  }

  /**
   * Initialize RingBuffer & RingBufferPaddingExecutor
   */
  private void initRingBuffer() {
    // initialize RingBuffer
    int bufferSize = ((int) bitsAllocator.getMaxSequence() + 1) << boostPower;
    this.ringBuffer = new RingBuffer(bufferSize, paddingFactor);
    log.info("Initialized ring buffer size:{}, paddingFactor:{}", bufferSize, paddingFactor);

    // initialize RingBufferPaddingExecutor
    boolean usingSchedule = (scheduleInterval != null);
    //jdk 1.8-->1.7
    //this.bufferPaddingExecutor = new BufferPaddingExecutor(ringBuffer, this::nextIdsForOneSecond, usingSchedule);
    BufferedUidProvider uidProvider = new BufferedUidProvider() {
      @Override
      public List<Long> provide(long momentInSecond) {
        return nextIdsForOneSecond(momentInSecond);
      }
    };
    this.bufferPaddingExecutor = new BufferPaddingExecutor(ringBuffer, uidProvider, usingSchedule);
    if (usingSchedule) {
      bufferPaddingExecutor.setScheduleInterval(scheduleInterval);
    }

    log.info("Initialized BufferPaddingExecutor. Using schdule:{}, interval:{}", usingSchedule, scheduleInterval);

    // set rejected put/take handle policy
    this.ringBuffer.setBufferPaddingExecutor(bufferPaddingExecutor);
    if (rejectedPutBufferHandler != null) {
      this.ringBuffer.setRejectedPutHandler(rejectedPutBufferHandler);
    }
    if (rejectedTakeBufferHandler != null) {
      this.ringBuffer.setRejectedTakeHandler(rejectedTakeBufferHandler);
    }

    // fill in all slots of the RingBuffer
    bufferPaddingExecutor.paddingBuffer();

    // start buffer padding threads
    bufferPaddingExecutor.start();
  }

  /**
   * Setters for spring property
   */
  public void setBoostPower(int boostPower) {
    Assert.isTrue(boostPower > 0, "Boost power must be positive!");
    this.boostPower = boostPower;
  }

  public void setRejectedPutBufferHandler(RejectedPutBufferHandler rejectedPutBufferHandler) {
    Assert.notNull(rejectedPutBufferHandler, "RejectedPutBufferHandler can't be null!");
    this.rejectedPutBufferHandler = rejectedPutBufferHandler;
  }

  public void setRejectedTakeBufferHandler(RejectedTakeBufferHandler rejectedTakeBufferHandler) {
    Assert.notNull(rejectedTakeBufferHandler, "RejectedTakeBufferHandler can't be null!");
    this.rejectedTakeBufferHandler = rejectedTakeBufferHandler;
  }

  public void setScheduleInterval(long scheduleInterval) {
    AssertUtil.isTrue(scheduleInterval > 0, "Schedule interval must positive!");
    this.scheduleInterval = scheduleInterval;
  }

}
